package cn.lsh.reducer;

import cn.lsh.conts.MyCounter;
import cn.lsh.writable.NetPage;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class PageRankReducer extends Reducer<Text, NetPage, Text, Text> {
	private final Text rkey = new Text();
	private final Text rval = new Text();

	@Override
	protected void reduce(Text key, Iterable<NetPage> values, Context context) throws IOException, InterruptedException {
		//输入数据样本也有两下两种：
		//key A 		value A		1.0 	B C 	0
		//key A			value A 	0.5     		1
		NetPage currPage = new NetPage();
		double trSum = 0.0;
		for (NetPage page : values) {
			if (page.getPageType() == 0) {
				//注意！！！reduce循环value的时候，不能将内部引用直接赋值给for循环外部用，
				// reduce的迭代器不会新建对象，而是改变的原来引用的值，所以这里需要深度拷贝对象
				// currPage = page;
				currPage.setPage(page.getPage());
				currPage.setPr(page.getPr());
				currPage.setRelationPage(page.getRelationPage());
				currPage.setPageType(page.getPageType());
			} else {
				//统计网站获得投票的权重总数值
				trSum += page.getPr();
			}
		}
		//获取网站总数
		int pageNum = context.getConfiguration().getInt("pageNum", 1);
		//获取阻尼系数，默认0.85
		double d = context.getConfiguration().getDouble("zuniNumber", 0.85);
		//根据公式计算新的pr值
		double newPr = (1 - d) / pageNum + d * trSum;
		//用新pr值和旧pr值计算差值
		double c = newPr - currPage.getPr();

		//放大1000倍，计算绝对值
		int j = (int) (c * 1000);
		//double强转int丢失小数位精度，0.8 -> 0，当j=0迭代停止，所以倍数
		j = Math.abs(j);

		//把值放到累加器中，计算pr差值所有网页的和
		context.getCounter(MyCounter.MY).setValue(j);

		rkey.set(currPage.getPage() + ":" + newPr);
		rval.set(currPage.arrToStr());
		context.write(rkey, rval);
	}
}
